package com.cctc.rds.raft.sdk.server;

import com.cctc.rds.nrpc.api.rpc.RpcServerRouter;
import com.cctc.rds.nrpc.api.server.config.RpcServerConfig;
import com.cctc.rds.nrpc.core.server.NettyRpcServer;
import com.cctc.rds.raft.core.RaftCluster;
import com.cctc.rds.raft.core.log.LogService;
import com.cctc.rds.raft.core.log.commit.statemachine.StateMachine;
import com.cctc.rds.raft.core.node.NodeIpPort;
import com.cctc.rds.raft.core.rpc.election.ElectionRpc;
import com.cctc.rds.raft.core.rpc.election.impl.ElectionRpcResponseImpl;
import com.cctc.rds.raft.core.rpc.log.LogRpc;
import com.cctc.rds.raft.core.rpc.log.impl.LogRpcResponseImpl;
import com.cctc.rds.raft.sdk.server.rpc.RaftRpcServerRouter;

import java.io.Closeable;
import java.util.Set;

public class RaftServer implements Closeable {
    private final int nodeId;
    private LogService logService;
    private StateMachine stateMachine;
    private Set<NodeIpPort> nodes;
    private NettyRpcServer server;

    public RaftServer(int nodeId, LogService logService, StateMachine stateMachine, Set<NodeIpPort> nodes) {
        this.nodeId = nodeId;
        this.logService = logService;
        this.stateMachine = stateMachine;
        this.nodes = nodes;
        initRaftCluster(nodes);
    }

    private void initRaftCluster(Set<NodeIpPort> nodes) {
        RaftCluster.getConfig().setNodeId(nodeId);
        RaftCluster.init(nodes);
    }

    public void start() {
        ElectionRpc electionRpc = ElectionRpcResponseImpl.builder().nodeId(nodeId).logService(logService).build();
        LogRpc logRpc = new LogRpcResponseImpl(nodeId, logService, stateMachine);
        RpcServerRouter router = new RaftRpcServerRouter(electionRpc, logRpc);
        NettyRpcServer server = new NettyRpcServer(router);
        NodeIpPort ipPort = RaftCluster.getNode(nodeId).getIpPort();
        RpcServerConfig serverConfig = RpcServerConfig.builder()
                .port(ipPort.getPort())
                .workerThreads(Runtime.getRuntime().availableProcessors())
                .idleStateMs(300)
                .build();
        server.start(serverConfig);
        this.server = server;
    }

    @Override
    public void close() {
        server.stop();
    }
}
